package com.atguigu.userprofile.app;


import com.atguigu.userprofile.common.bean.TagInfo;
import com.atguigu.userprofile.common.constants.ConstCode;
import com.atguigu.userprofile.common.dao.TagInfoDAO;
import com.atguigu.userprofile.common.util.MyClickhouseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TaskBitmapApp {


    //1 标签列表 mysql
    //
    //2  建表 bitmap表
    //  手动建？ 自动建 ？   每天建？  永久建一次？因表结构不会随着标签变化而变化  永久建1次即可
    //  一共建四张表， 按照不同的标签值类型 建表
    //
    //
    //3    insert  select    在clickhouse中完成
    public static void main(String[] args) {

        //本程序并没有真正使用spark进行分布式计算，之所以还引入spark 环境 主要是为了所以程序统一提交统一调度 比较方便。
       //  System.setProperty("hadoop.home.dir", "d:\\hadoop");//打包时注释掉
        SparkConf sparkConf = new SparkConf().setAppName("task_bitmap_app");//.setMaster("local[*]");
        SparkContext sparkContext = new SparkContext(sparkConf);


        String taskId = args[0];
        String busiDate = args[1];
        //1 标签列表 mysql

        List<TagInfo> tagInfoList = TagInfoDAO.getTagInfoListWithOn();

        List<TagInfo> tagInfoListForString =new ArrayList<>();
        List<TagInfo> tagInfoListForLong =new ArrayList<>();
        List<TagInfo> tagInfoListForDecimal =new ArrayList<>();
        List<TagInfo> tagInfoListForDate=new ArrayList<>();
        //2  根据标签值把标签分成四份
        for (TagInfo tagInfo : tagInfoList) {
            if(tagInfo.getTagValueType().equals(ConstCode.TAG_VALUE_TYPE_STRING)){
                tagInfoListForString.add(tagInfo);
            }else if (tagInfo.getTagValueType().equals(ConstCode.TAG_VALUE_TYPE_LONG)) {
                tagInfoListForLong.add(tagInfo);
            }else if (tagInfo.getTagValueType().equals(ConstCode.TAG_VALUE_TYPE_DECIMAL)) {
                tagInfoListForDecimal.add(tagInfo);
            }else if (tagInfo.getTagValueType().equals(ConstCode.TAG_VALUE_TYPE_DATE)) {
                tagInfoListForDate.add(tagInfo);
            }
        }


        //3    insert  select    在clickhouse中完成
        insertBitmap(tagInfoListForString,  "user_tag_value_string",  busiDate);
        insertBitmap(tagInfoListForLong,  "user_tag_value_long",  busiDate);
        insertBitmap(tagInfoListForDecimal,  "user_tag_value_decimal",  busiDate);
        insertBitmap(tagInfoListForDate,  "user_tag_value_date",  busiDate);



    }

    //不同的标签插入不同的表
    private static void   insertBitmap(List<TagInfo> tagInfoList,String targetTableName,String busiDate){

        //insert into  user_tag_value_string
        //select    tv.1,tv.2 , bitmapToArray(groupBitmapState(cast(uid as UInt64)  ) ) ,'2020-06-14'from
        //(
        //  select uid ,
        //  arrayJoin([('tg_person_base_gender',tg_person_base_gender),('tg_person_base_agegroup',tg_person_base_agegroup)] ) tv
        //  from up_tag_merge_20200614
        // ) tt
        //group by tv.1,tv.2
        if(tagInfoList.size()>0) {

            String deleteSQL= "alter table "+targetTableName+" delete where dt='"+busiDate+"'";

            System.out.println(deleteSQL);
            MyClickhouseUtil.executeSql(deleteSQL);

            List<String> tagCodeList = tagInfoList.stream().map(tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "'," + tagInfo.getTagCode().toLowerCase() + ")").collect(Collectors.toList());
            String tagCodeSQL = StringUtils.join(tagCodeList, ",");

            String insertSQL = "  insert into  " + targetTableName +
                    "        select    tv.1, if(tv.2='','0',tv.2) ,  groupBitmapState(cast(uid as UInt64)  )   ,'" + busiDate + "'" +
                    "        from (" +
                    "       select uid ,\n" +
                    "         arrayJoin([" + tagCodeSQL + "] ) tv\n" +
                    "           from up_tag_merge_" + busiDate.replace("-", "") +
                    "          ) tt\n" +
                    "        group by tv.1,tv.2";

            System.out.println(insertSQL);
            MyClickhouseUtil.executeSql(insertSQL); //dr? ex?  // 用到spark分布式计算了吗？


        }

    }
}
